%% copy riak-0.14.2 source to learn dynamo implementation and erlang
%% author: lichuang
%% copy source from riak_core_util.erl, riak_kv/src/riak.erl, riak_kv_util.erl
-module(edynamo_util).
-include("riak_kv_vnode.hrl").

-define(MAX_KEY_SIZE, 65536).

-export([moment/0,
         client_connect/1,client_connect/2,
         integer_to_list/2,
         compare_dates/2,
         try_cast/3,
         fallback/4,
         expand_rw_value/4,
         normalize_rw_value/2,
         mkclientid/1,
         chash_key/1,
         chash_std_keyfun/1,
         make_request/2,
         join/1,
         str_to_node/1,
         is_x_deleted/1, obj_not_deleted/1,
         local_client/0, local_client/1]).

%% @spec is_x_deleted(edynamo_object:edynamo_object()) -> boolean()
%% @doc 'true' if all contents of the input object are marked
%%      as deleted; 'false' otherwise
%% @equiv obj_not_deleted(Obj) == undefined
is_x_deleted(Obj) ->
  case obj_not_deleted(Obj) of
    undefined -> true;
    _ -> false
  end.

%% @spec obj_not_deleted(edynamo_object:edynamo_object()) ->
%%          undefined|edynamo_object:edynamo_object()
%% @doc Determine whether all contents of an object are marked as
%%      deleted.  Return is the atom 'undefined' if all contents
%%      are marked deleted, or the input Obj if any of them are not.
obj_not_deleted(Obj) ->
  case [{M, V} || {M, V} <- edynamo_object:get_contents(Obj),
                  dict:is_key(<<"X-Riak-Deleted">>, M) =:= false] of
    [] -> undefined;
    _  -> Obj
  end.      

%% @spec chash_key(BKey :: edynamo_object:bkey()) -> chash:index()
%% @doc Create a binary used for determining replica placement.
chash_key({Bucket, Key}) ->
  BucketProps = edynamo_bucket:get_bucket(Bucket),
  {chash_keyfun, {M, F}} = proplists:lookup(chash_keyfun, BucketProps),
  M:F({Bucket, Key}).

%% @spec chash_std_keyfun(BKey :: edynamo_object:bkey()) -> chash:index()
%% @doc Default object/ring hashing fun, direct passthrough of bkey.
chash_std_keyfun({Bucket, Key}) ->
  chash:key_of({Bucket, Key}).

%% @doc Get the current "moment".Current implementation is the
%%	number of seconds from year 0 to now, universal time, in
%%	the gregorian calendar
moment() ->
  calendar:datetime_to_gregorian_seconds(calendar:universal_time()).

%% @spec local_client() -> {ok, Client::riak_client()}
%% @equiv local_client(undefined)
local_client() ->
  local_client(undefined).

%% @spec local_client(binary()|undefined) -> {ok, Client::riak_client()}
%% @doc When you want a client for use on a running Riak node.
%%      ClientId should be a 32-bit binary.  If it is not, a
%%      32-bit binary will be created from ClientId by phash2/1.
%%      If ClientId is the atom 'undefined', a random ClientId will
%%      be chosen.
local_client(ClientId) ->
  client_connect(node(), ClientId).

%% @spec client_connect(Node :: node())
%%        -> {ok, Client :: riak_client()} | {error, timeout}
%% @equiv client_connect(Node, undefined)
client_connect(Node) ->
  client_connect(Node, undefined).

%% @spec client_connect(node(), binary()|undefined)
%%         -> {ok, Client :: riak_client} | {error, timeout}
%% @doc The usual way to get a client.  Timeout often means either a bad
%%      cookie or a poorly-connected distributed erlang network.
%%      ClientId should be a 32-bit binary.  If it is not, a
%%      32-bit binary will be created from ClientId by phash2/1.
%%      If ClientId is the atom 'undefined', a random ClientId will
%%      be chosen.
client_connect(Node, ClientId = <<_:32>>) ->
  % Make sure we can reach this node
  case net_adm:ping(Node) of
    pang -> {error, {could_not_reach_node, Node}};
    pong -> {ok, edynamo_client:new(Node, ClientId)}
  end;
client_connect(Node, undefined) ->
  client_connect(Node, edynamo_util:mkclientid(Node));
client_connect(Node, Other) ->
  client_connect(Node, <<(erlang:phash2(Other))>>).

%% @spec mkclientid(RemoteNode :: term()) -> ClientID :: list()
%% @doc Create a unique-enough id for vclock clients.
mkclientid(RemoteNode) ->
  {{Y,Mo,D},{H,Mi,S}} = erlang:universaltime(),
  {_, _, NowPart} = now(),
  Id = erlang:phash2([Y, Mo, D, H, Mi, S, node(), RemoteNode, NowPart]),
  <<Id:32>>.

expand_rw_value(Type, default, BucketProps, N) ->
  normalize_rw_value(get_default_rw_val(Type, BucketProps), N);
expand_rw_value(_Type, Val, _BucketProps, N) ->
  normalize_rw_value(Val, N).

normalize_rw_value(RW, _N) when is_integer(RW) -> RW;
normalize_rw_value(RW, N) when is_binary(RW) ->
    normalize_rw_value(binary_to_atom(RW, utf8), N);
normalize_rw_value(one, _N) -> 1;
normalize_rw_value(quorum, N) -> erlang:trunc((N/2)+1);
normalize_rw_value(all, N) -> N.

get_default_rw_val(Type, BucketProps) ->
  {ok, DefaultProps} = application:get_env(edynamo, default_bucket_props),
  case {proplists:get_value(Type, BucketProps),
        proplists:get_value(Type, DefaultProps)} of
      {undefined, Val} -> Val;
      {Val, undefined} -> Val;
      {Val1, _Val2}    -> Val1
  end.        

%% @spec fallback(term(), term(), [{Index :: term(), Node :: node()}],
%%                [{any(), Fallback :: node()}]) ->
%%         [{Index :: term(), Node :: node(), Fallback :: node()}]
%% @doc Cast {Cmd, {Index,Node}, Msg} at a node in the Fallbacks list
%%      for each node in the Pangs list.  Pangs should have come
%%      from the second element of the response tuple of a call to
%%      try_cast/3.
%%      Used in edynamo_put_fsm and edynamo_get_fsm
fallback(Cmd, UpNodes, Pangs, Fallbacks) ->
  fallback(Cmd, UpNodes, Pangs, Fallbacks, []).
fallback(_Cmd, _UpNodes, [], _Fallbacks, Sent) ->
  Sent;
fallback(_Cmd, _UpNodes, _Pangs, [], Sent) ->
  Sent;
fallback(Cmd, UpNodes, [{Index, Node} | Pangs], [{_, FN} | Fallbacks], Sent) ->
  case lists:member(FN, UpNodes) of
    false ->
      fallback(Cmd, UpNodes, [{Index, Node} | Pangs], Fallbacks, Sent);
    true  ->
      gen_server:cast({edynamo_vnode_master, FN}, make_request(Cmd, Index)),
      fallback(Cmd, UpNodes, Pangs, Fallbacks, [{Index, Node, FN} | Sent])
  end.

%% @spec try_cast(term(), [node()], [{Index :: term(), Node :: node()}]) ->
%%          {[{Index :: term(), Node :: node(), Node :: node()}],
%%           [{Index :: term(), Node :: node()}]}
%% @doc Cast {Cmd, {Index,Node}, Msg} at riak_kv_vnode_master on Node
%%      if Node is in UpNodes.  The list of successful casts is the
%%      first element of the return tuple, and the list of unavailable
%%      nodes is the second element.  Used in riak_kv_put_fsm and riak_kv_get_fsm.
try_cast(Msg, UpNodes, Targets) ->
  try_cast(Msg, UpNodes, Targets, [], []).
try_cast(_Msg, _UpNodes, [], Sent, Pangs) ->
  {Sent, Pangs};
try_cast(Msg, UpNodes, [{Index, Node} | Targets], Sent, Pangs) ->
  case lists:member(Node, UpNodes) of
    false ->
      try_cast(Msg, UpNodes, Targets, Sent, [{Index, Node} | Pangs]);
    true  ->
      gen_server:cast({edynamo_kv_vnode_master, Node}, make_request(Msg, Index)),
      try_cast(Msg, UpNodes, Targets, [{Index, Node, Node} | Sent], Pangs)
  end.

-spec make_request(vnode_req(), partition()) -> #riak_vnode_req_v1{}.
make_request(Request, Index) ->
  edynamo_vnode_master:make_request(Request, {fsm, undefined, self()}, Index).

%% @spec integer_to_list(Integer :: integer(), Base :: integer()) ->
%%          string()
%% @doc Convert an integer to its string representation in the given
%%      base.  Bases 2-62 are supported.
integer_to_list(I, 10) ->
  erlang:integer_to_list(I);
integer_to_list(I, Base)
  when is_integer(I), is_integer(Base), Base >= 2, Base =< 1+$Z-$A+10+1+$z-$a ->
    if I < 0 ->
        [$-|integer_to_list(-I, Base, [])];
      true ->
        integer_to_list(I, Base, [])
    end;
integer_to_list(I, Base) ->
  erlang:error(badarg, [I]).
%% @spec integer_to_list(integer(), integer(), string()) -> string()
integer_to_list(I0, Base, R0) ->
  D = I0 rem Base,
  I1 = I0 div Base,
  R1 = if D >= 36 ->
      [D-36+$a|R0];
    D >= 10 ->
      [D-10+$A|R0];
    true ->
      [D+$0|R0]
  end,
  if I1 =:= 0 ->
      R1;
    true ->
      integer_to_list(I1, Base, R1)
  end.

%% @spec compare_dates(string(), string()) -> boolean()
%% @doc Compare two RFC1123 date strings or two now() tuples (or one
%%      of each).  Return true if date A is later than date B.
compare_dates(A={_,_,_}, B={_,_,_}) ->
  %% assume 3-tuples are now() times
  A > B;
compare_dates(A, B) when is_list(A) ->
  %% assume lists are rfc1123 date strings
  compare_dates(rfc1123_to_now(A), B);
compare_dates(A, B) when is_list(B) ->
  compare_dates(A, rfc1123_to_now(B)).

%% 719528 days from Jan 1, 0 to Jan 1, 1970
%%  *86400 seconds/day
-define(SEC_TO_EPOCH, 62167219200).

rfc1123_to_now(String) when is_list(String) ->
  GSec = calendar:datetime_to_gregorian_seconds(
    httpd_util:convert_request_date(String)),
  ESec = GSec-?SEC_TO_EPOCH,
  Sec = ESec rem 1000000,
  MSec = ESec div 1000000,
  {MSec, Sec, 0}.

%%
%% @doc Join the ring found on the specified remote node
%%
join(NodeStr) when is_list(NodeStr) ->
  join(edynamo_util:str_to_node(NodeStr));
join(Node) when is_atom(Node) ->
  {ok, OurRingSize} = application:get_env(edynamo, ring_creation_size),
  case net_adm:ping(Node) of
    pong ->
      case rpc:call(Node,
                    application,
                    get_env,
                    [edynamo, ring_creation_size]) of
        {ok, OurRingSize} ->
          edynamo_gossip:send_ring(Node, node());
        _ ->
          {error, different_ring_sizes}
      end;
    pang ->
      {error, not_reachable}
  end.      

str_to_node(Node) when is_atom(Node) ->
  str_to_node(atom_to_list(Node));
str_to_node(NodeStr) ->
  case string:tokens(NodeStr, "@") of
    [NodeName] ->
      %% Node name only; no host name. If the local node has a hostname,
      %% append it
      case node_hostname() of
        [] ->
          list_to_atom(NodeName);
        Hostname ->
          list_to_atom(NodeName ++ "@" ++ Hostname)
      end;
    _ ->
      list_to_atom(NodeStr)
  end.      

node_hostname() ->
  NodeStr = atom_to_list(node()),
  case string:tokens(NodeStr, "@") of
    [_NodeName, Hostname] ->
      Hostname;
    _ ->
      []
  end.      
